package com.iqiyi.pps.epg.core.service.vod;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.iqiyi.kiwi.utils.Configuration;
import com.iqiyi.pps.epg.api.model.web.query.TFilter;
import com.iqiyi.pps.epg.core.async.AsyncService;
import com.iqiyi.pps.epg.core.async.HeartBeatCenter;
import com.iqiyi.pps.epg.core.async.IAsyncTask;
import com.iqiyi.pps.epg.core.async.IHeartBeat;
import com.iqiyi.pps.epg.core.model.channel.ChannelConfig;
import com.iqiyi.pps.epg.core.model.task.ProgramSimpleTask;
import com.iqiyi.pps.epg.core.model.task.ProgramSimpleTaskType;
import com.iqiyi.pps.epg.core.model.task.SimpleTaskStatus;
import com.iqiyi.pps.epg.core.model.vod.VodRecord;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.service.task.ProgramSimpleTaskService;
import com.iqiyi.pps.epg.core.utils.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.util.*;


/**
 * Created by karkarrotxu on 2017/12/15.
 */
@Lazy(false)
@Service
public class VodAsyncServiceImpl extends AsyncService implements InitializingBean, IHeartBeat, IVodAsyncService {

    private static final Logger logger = LoggerFactory.getLogger(VodAsyncServiceImpl.class);
    private final static int HEARTBEAT_DELAY = 10 * 1000; //
    private long heartBeatTime = 0;
    public static String ASSIGN_WORK_IP = "10.153.132.18";
    public static final int RECORD_PREVIEW_SECOND = Configuration.getInt("record.preview.second");
    @Autowired
    private ProgramSimpleTaskService simpleTaskService;

    @Autowired
    private ChannelService channelService;

    @Autowired
    private IVodService vodService;

    public void afterPropertiesSet() throws Exception {
        logger.info("VodAsyncServiceImpl afterPropertiesSet");
        HeartBeatCenter.add(this);
    }

    protected boolean queryEnable() {
        return getEnable("vod");
    }

    @Override
    public void tick(long now) {
        if (now - heartBeatTime >= HEARTBEAT_DELAY) {
            heartBeatTime = now;
            logger.info("triggerOnce tick");
            triggerStopOnce();
            triggerStartOnce();
        }
    }

    public void triggerStartOnce() {
        logger.info("NOTIFY_VOD_RECORD_START tick");
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                logger.info("NOTIFY_VOD_RECORD_START query");
                executeSimpleTaskList(ProgramSimpleTaskType.NOTIFY_VOD_RECORD_START.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    public void triggerStopOnce() {
        logger.info("NOTIFY_VOD_RECORD_END tick");
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                logger.info("NOTIFY_VOD_RECORD_END query");
                executeSimpleTaskList(ProgramSimpleTaskType.NOTIFY_VOD_RECORD_END.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    @Override
    public void addRecordTask(VodRecord vodRecord) {
        interruptTask(vodRecord.getId());
        String currentTime = DateTimeUtil.transLongFormat(System.currentTimeMillis());
        boolean hasEnd = vodRecord.getEndTime() != null;
        int compareStart = currentTime.compareTo(vodRecord.getStartTime());
        int compareEnd = 0;
        if (hasEnd) {
            compareEnd = currentTime.compareTo(vodRecord.getEndTime());
            if (compareEnd >= 0) {
                logger.info("recordTask {},old time skip task", vodRecord.getId());
                return;
            }
        } else {
            compareEnd = -1;
        }
        if (compareEnd < 0 && compareStart > 0) {
            if (hasEnd) {
                addRecordStopTask(vodRecord);
            }
        } else if (compareStart < 0) {
            addRecordStartTask(vodRecord);
        }

    }

    private void addRecordStartTask(VodRecord entity) {
        ProgramSimpleTask startTask = new ProgramSimpleTask();
        startTask.setServerIp(RequestUtils.getLoadlIpByOS());
        startTask.setWorkIp(isProductionEnv() ? ASSIGN_WORK_IP : startTask.getServerIp());
        startTask.setTaskType(ProgramSimpleTaskType.NOTIFY_VOD_RECORD_START.getTypeIndex());
        startTask.setEntityId(entity.getId());
        startTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        Timestamp timestamp = Timestamp.valueOf(entity.getStartTime());
        timestamp.setTime(timestamp.getTime() - DateTimeUtil.SECOND * RECORD_PREVIEW_SECOND);
        startTask.setStartUpTime(timestamp);
        startTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        if (entity.getEndTime() == null) {
            startTask.setExpireTime(new Timestamp(startTask.getStartUpTime().getTime() + 5 * DateTimeUtil.MINUTE_TIME));
        } else {
            startTask.setExpireTime(Timestamp.valueOf(entity.getEndTime()));
        }
        simpleTaskService.saveTask(startTask);
        triggerStartOnce();
    }


    private void addRecordStopTask(VodRecord entity) {
        if (entity.getEndTime() == null) {
            return;
        }
        ProgramSimpleTask stopTask = new ProgramSimpleTask();
        stopTask.setServerIp(RequestUtils.getLoadlIpByOS());
        stopTask.setWorkIp(isProductionEnv() ? ASSIGN_WORK_IP : stopTask.getServerIp());
        stopTask.setTaskType(ProgramSimpleTaskType.NOTIFY_VOD_RECORD_END.getTypeIndex());
        stopTask.setEntityId(entity.getId());
        stopTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        stopTask.setStartUpTime(Timestamp.valueOf(entity.getEndTime()));
        stopTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        stopTask.setExpireTime(new Timestamp(stopTask.getStartUpTime().getTime() + 2 * DateTimeUtil.MINUTE_TIME));
        simpleTaskService.saveTask(stopTask);
        triggerStopOnce();
    }

    @Override
    public void interruptTask(long recordId) {
        int effectNum = simpleTaskService.interruptTask(recordId, ProgramSimpleTaskType.NOTIFY_VOD_RECORD_START.getTypeIndex());
        logger.info("[effectNum]NOTIFY_VOD_RECORD_START rId {},effectNum={}:", recordId, effectNum);
        effectNum = simpleTaskService.interruptTask(recordId, ProgramSimpleTaskType.NOTIFY_VOD_RECORD_END.getTypeIndex());
        logger.info("[effectNum]NOTIFY_VOD_RECORD_END rId {},effectNum={}:", recordId, effectNum);
    }

    @Override
    public void interruptTaskByChannel(int channelId) {
        Map<String, String> params = new HashMap();
        params.put("EQI_status", "1");
        params.put("EQI_channelId", channelId + "");
        TFilter tFilter = new TFilter();
        tFilter.setFilters(params);
        List<VodRecord> list = vodService.findRecordList(tFilter);
        logger.info("[interruptTaskByChannel][size={}]", list.size());
        if (list.size() > 0) {
            for (VodRecord entity : list) {
                interruptTask(entity.getId());
            }
        }
    }

    protected void executeSimpleTaskList(int type, int status, boolean needOverride) {
        logger.info("[executeSimpleTaskList] execute type:{},status={}", type, status);
        try {
            List<ProgramSimpleTask> resultList = simpleTaskService.findList(type, status);
            if (resultList == null || resultList.size() == 0) {
                logger.info("[result empty] resultList == null :" + (resultList == null));
                return;
            }
            logger.info("[executeSimpleTaskList] execute size:" + resultList.size());
            long currentTime = System.currentTimeMillis();
            if (needOverride) {
                Collections.reverse(resultList);
            }
            List<ProgramSimpleTask> updateList = new ArrayList<>(2048);
            Map<Long, ProgramSimpleTask> checkMap = new HashMap<>(2048);
            List<ProgramSimpleTask> waitToDealList = new ArrayList<>(2048);
            for (ProgramSimpleTask task : resultList) {
                if (currentTime > task.getExpireTime().getTime()) {
                    task.setTaskStatus(SimpleTaskStatus.EXPIRED.getStatusIndex());
                    updateList.add(task);
                } else {
                    if (needOverride) {
                        if (!checkMap.containsKey(task.getEntityId())) {
                            checkMap.put(task.getEntityId(), task);
                        } else {
                            task.setTaskStatus(SimpleTaskStatus.OVERRIDE.getStatusIndex());
                            updateList.add(task);
                        }
                    } else {
                        waitToDealList.add(task);
                    }
                }
            }
            logger.info("[executeSimpleTaskList] overtime size:" + updateList);
            if (updateList.size() > 0) {
                simpleTaskService.batchUpdateList(updateList);
            }

            if (checkMap.size() > 0) {
                waitToDealList.addAll(checkMap.values());
            }

            if (waitToDealList.size() > 0) {
                updateList.clear();
                for (ProgramSimpleTask task : waitToDealList) {
                    if (executeSimpleTask(task)) {
                        updateList.add(task);
                    }
                }
                if (updateList.size() > 0) {
                    simpleTaskService.batchUpdateList(updateList);
                }
            }
        } catch (Exception e) {
            logger.error("[executeSimpleTaskList] error", e);
        }
        logger.info("[executeSimpleTaskList] execute end");
    }

    protected boolean executeSimpleTask(ProgramSimpleTask simpleTask) {
        ProgramSimpleTaskType type = ProgramSimpleTaskType.getTypeByIndex(simpleTask.getTaskType());
        boolean flag = false;
        switch (type) {
            case NOTIFY_VOD_RECORD_START:
                flag = doRecordStart(simpleTask);
                break;
            case NOTIFY_VOD_RECORD_END:
                flag = doRecordStop(simpleTask);
                break;
        }
        if (flag) {
            simpleTask.setTaskStatus(SimpleTaskStatus.COMPLETE.getStatusIndex());
            simpleTask.setCompleteTime(new Timestamp(System.currentTimeMillis()));
        }
        return flag;
    }

    private void writeVodLog(int channelId, int type, String descr) {
        WriteSystemLogUtil.getInstance().writeVodLog(channelId, type, Constants.SYSTEM_USER_SIGN, descr);
    }

    private boolean canUseAlert(VodRecord vodRecord) {
        return vodRecord.getProjectWorkers() != null && vodRecord.getProjectWorkers().length() > 2 && vodRecord.getNotifyType() > 0;
    }

    private boolean doRecordStart(ProgramSimpleTask simpleTask) {
        VodRecord vodRecord = vodService.findRecordById(simpleTask.getEntityId());
        JSONObject entity = ApiVodUtil.queryChannelById(vodRecord.getChannelId());
        int analyzeValue = ApiVodUtil.analyzeQueryJson(entity);
        if (analyzeValue != ApiVodUtil.ANALYZE_RESULT_SUC) {
            dealErrorResponse(entity, simpleTask, vodRecord, Constants.VodLogType.VOD_AUTO_RECORD_START.getIndex(), "自动开始收录失败");
            return false;
        }
        JSONObject element = entity.getJSONObject("response").getJSONArray("list").getJSONObject(0);
        Integer status = element.getInteger("channel_status");
        if (status == 1) {
            logger.info("[{},{}] is in record ,skip", simpleTask.getEntityId(), simpleTask.getId());
            writeVodLog(vodRecord.getChannelId(), Constants.VodLogType.VOD_AUTO_RECORD_START.getIndex(), "task(" + simpleTask.getId() + ") skip start | case channel is in record");
            return true;
        }
        JSONObject retJson = ApiVodUtil.doStartRecord(element);
        writeVodLog(vodRecord.getChannelId(), Constants.VodLogType.VOD_AUTO_RECORD_START.getIndex(), "task(" + simpleTask.getId() + ") notify record | ret =" + retJson);
        boolean retValue = ApiVodUtil.analyzeResponseJson(retJson) == ApiVodUtil.ANALYZE_RESULT_SUC;
        if (canUseAlert(vodRecord)) {
            JSONObject content = new JSONObject();
            content.put("content", element.getString("channel_name") + "自动开始收录" + (retValue ? "成功" : "失败"));
            ApiAlertPlatFormUtil.doSend(ApiAlertPlatFormUtil.TOPIC_ID_LIVE_RECORD, ApiAlertPlatFormUtil.SECRET_KEY_EPG, content.toJSONString(), getWorkers(vodRecord), vodRecord.getNotifyType());
        }
        if (retValue) {
            addRecordStopTask(vodRecord);
        } else {// send on error
            doTaskError(simpleTask);
            sendMsg("频道id:[" + vodRecord.getChannelId() + "]" + element.getString("channel_name") + ",自动开始收录失败");
        }
        return retValue;
    }

    private String getWorkers(VodRecord vodRecord) {
        JSONArray jsonArray = JSONArray.parseArray(vodRecord.getProjectWorkers());
        Iterator<Object> t = jsonArray.iterator();
        JSONObject entity = null;
        StringBuilder sb = new StringBuilder();
        while (t.hasNext()) {
            entity = (JSONObject) t.next();
            sb.append(entity.getString("id").split("@")[0]);
            sb.append(",");
        }
        if (sb.length() > 0) {
            sb.deleteCharAt(sb.length() - 1);
        }
        return sb.toString();
    }

    private void sendMsg(String content) {
        SendMsgUtils.sendMsg(getTargets(), content, "【epg】", "warn");
    }

    private String getTargets() {
        List<ChannelConfig> configs = channelService.getConfig(ChannelConfig.ZHIBO_TEAM);
        StringBuffer sb = new StringBuffer();
        int size = configs.size();
        for (int i = 0; i < size; i++) {
            if (i > 0) {
                sb.append(",");
            }
            sb.append(configs.get(i).getVal());
        }
        return sb.toString();
    }

    private void dealErrorResponse(JSONObject entity, ProgramSimpleTask simpleTask, VodRecord vodRecord, int type, String erroMsg) {
        doTaskError(simpleTask);
        writeVodLog(vodRecord.getChannelId(), type, "task(" + simpleTask.getId() + ") executed error | ret=" + entity);
        sendMsg("[cid=" + vodRecord.getChannelId() + "][sid=" + simpleTask.getId() + "] ," + erroMsg);
        if (canUseAlert(vodRecord)) {
            JSONObject content = new JSONObject();
            content.put("content", "[cid=" + vodRecord.getChannelId() + "]" + erroMsg);
            ApiAlertPlatFormUtil.doSend(ApiAlertPlatFormUtil.TOPIC_ID_LIVE_RECORD, ApiAlertPlatFormUtil.SECRET_KEY_EPG, content.toJSONString(), getWorkers(vodRecord), vodRecord.getNotifyType());
        }
    }

    private boolean doRecordStop(ProgramSimpleTask simpleTask) {
        VodRecord vodRecord = vodService.findRecordById(simpleTask.getEntityId());
        JSONObject entity = ApiVodUtil.queryChannelById(vodRecord.getChannelId());
        int analyzeValue = ApiVodUtil.analyzeQueryJson(entity);
        if (analyzeValue != ApiVodUtil.ANALYZE_RESULT_SUC) {
            dealErrorResponse(entity, simpleTask, vodRecord, Constants.VodLogType.VOD_AUTO_RECORD_STOP.getIndex(), "自动停止收录失败");
            return false;
        }
        JSONObject element = entity.getJSONObject("response").getJSONArray("list").getJSONObject(0);
        Integer status = element.getInteger("channel_status");
        if (status == 0) {
            logger.info("[{},{}] is stop ,skip", simpleTask.getEntityId(), simpleTask.getId());
            writeVodLog(vodRecord.getChannelId(), Constants.VodLogType.VOD_AUTO_RECORD_STOP.getIndex(), "task(" + simpleTask.getId() + ") skip stop | case channel is stop");
            return true;
        }
        JSONObject retJson = ApiVodUtil.doStopRecord(element);
        writeVodLog(vodRecord.getChannelId(), Constants.VodLogType.VOD_AUTO_RECORD_STOP.getIndex(), "task(" + simpleTask.getId() + ") notify stop | ret =" + retJson);
        boolean retValue = ApiVodUtil.analyzeResponseJson(retJson) == ApiVodUtil.ANALYZE_RESULT_SUC;
        if (canUseAlert(vodRecord)) {
            JSONObject content = new JSONObject();
            content.put("content", element.getString("channel_name") + "自动停止收录" + (retValue ? "成功" : "失败"));
            ApiAlertPlatFormUtil.doSend(ApiAlertPlatFormUtil.TOPIC_ID_LIVE_RECORD, ApiAlertPlatFormUtil.SECRET_KEY_EPG, content.toJSONString(), getWorkers(vodRecord), vodRecord.getNotifyType());
        }

        if (!retValue) {// send on error
            doTaskError(simpleTask);
            sendMsg("频道id:[" + vodRecord.getChannelId() + "]" + element.getString("channel_name") + "，自动停止收录失败");
        }
        return retValue;
    }

    private void doTaskError(ProgramSimpleTask simpleTask) {
        logger.info("simpleTask error id:{}", simpleTask.getId());
        simpleTask.setTaskStatus(SimpleTaskStatus.ERROR.getStatusIndex());
        simpleTaskService.saveTask(simpleTask);
    }
}
